package com.xxl.job.admin.core.thread;

import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.model.XxlJobGroup;
import com.xxl.job.admin.core.model.XxlJobRegistry;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.enums.RegistryConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.*;

/**
 * job registry instance
 *
 * @author xuxueli 2016-10-02 19:10:24
 */
public class JobRegistryHelper {

  private static Logger logger = LoggerFactory.getLogger(JobRegistryHelper.class);

  private static JobRegistryHelper instance = new JobRegistryHelper();

  public static JobRegistryHelper getInstance() {
    return instance;
  }

  private ThreadPoolExecutor registryOrRemoveThreadPool = null;
  private Thread registryMonitorThread;
  private volatile boolean toStop = false;

  public void start() {

    // for registry or remove
    registryOrRemoveThreadPool = new ThreadPoolExecutor(
        2,
        10,
        30L,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(2000),
        new ThreadFactory() {
          @Override
          public Thread newThread(Runnable r) {
            return new Thread(r,
                "xxl-job, admin JobRegistryMonitorHelper-registryOrRemoveThreadPool-"
                    + r.hashCode());
          }
        },
        new RejectedExecutionHandler() {
          @Override
          public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            r.run();
            logger.warn(
                ">>>>>>>>>>> xxl-job, registry or remove too fast, match threadpool rejected handler(run now).");
          }
        });

    // for monitor
    registryMonitorThread = new Thread(new Runnable() {
      @Override
      public void run() {
        while (!toStop) {
          try {
            // auto registry group
            List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao()
                .findByAddressType(0);
            if (groupList != null && !groupList.isEmpty()) {

              // remove dead address (admin/executor)
              List<Integer> ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
                  .findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
              if (ids != null && ids.size() > 0) {
                XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids);
              }

              // fresh online address (admin/executor)
              HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
              List<XxlJobRegistry> list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
                  .findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
              if (list != null) {
                for (XxlJobRegistry item : list) {
                  if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
                    String appname = item.getRegistryKey();
                    List<String> registryList = appAddressMap.get(appname);
                    if (registryList == null) {
                      registryList = new ArrayList<String>();
                    }

                    if (!registryList.contains(item.getRegistryValue())) {
                      registryList.add(item.getRegistryValue());
                    }
                    appAddressMap.put(appname, registryList);
                  }
                }
              }

              // fresh group address
              for (XxlJobGroup group : groupList) {
                List<String> registryList = appAddressMap.get(group.getAppname());
                String addressListStr = null;
                if (registryList != null && !registryList.isEmpty()) {
                  Collections.sort(registryList);
                  StringBuilder addressListSB = new StringBuilder();
                  for (String item : registryList) {
                    addressListSB.append(item).append(",");
                  }
                  addressListStr = addressListSB.toString();
                  addressListStr = addressListStr.substring(0, addressListStr.length() - 1);
                }
                group.setAddressList(addressListStr);
                group.setUpdateTime(new Date());

                XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group);
              }
            }
          } catch (Exception e) {
            if (!toStop) {
              logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
            }
          }
          try {
            TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
          } catch (InterruptedException e) {
            if (!toStop) {
              logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
            }
          }
        }
        logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
      }
    });
    registryMonitorThread.setDaemon(true);
    registryMonitorThread.setName("xxl-job, admin JobRegistryMonitorHelper-registryMonitorThread");
    registryMonitorThread.start();
  }

  public void toStop() {
    toStop = true;

    // stop registryOrRemoveThreadPool
    registryOrRemoveThreadPool.shutdownNow();

    // stop monitir (interrupt and wait)
    registryMonitorThread.interrupt();
    try {
      registryMonitorThread.join();
    } catch (InterruptedException e) {
      logger.error(e.getMessage(), e);
    }
  }

  // ---------------------- helper ----------------------

  public ReturnT<String> registry(RegistryParam registryParam) {

    // valid
    if (!StringUtils.hasText(registryParam.getRegistryGroup())
        || !StringUtils.hasText(registryParam.getRegistryKey())
        || !StringUtils.hasText(registryParam.getRegistryValue())) {
      return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
    }

    // async execute
    registryOrRemoveThreadPool.execute(new Runnable() {
      @Override
      public void run() {
        int ret = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
            .registryUpdate(registryParam.getRegistryGroup(), registryParam.getRegistryKey(),
                registryParam.getRegistryValue(), new Date());
        if (ret < 1) {
          XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
              .registrySave(registryParam.getRegistryGroup(), registryParam.getRegistryKey(),
                  registryParam.getRegistryValue(), new Date());

          // fresh
          freshGroupRegistryInfo(registryParam);
        }
      }
    });

    return ReturnT.SUCCESS;
  }

  public ReturnT<String> registryRemove(RegistryParam registryParam) {

    // valid
    if (!StringUtils.hasText(registryParam.getRegistryGroup())
        || !StringUtils.hasText(registryParam.getRegistryKey())
        || !StringUtils.hasText(registryParam.getRegistryValue())) {
      return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
    }

    // async execute
    registryOrRemoveThreadPool.execute(new Runnable() {
      @Override
      public void run() {
        int ret = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
            .registryDelete(registryParam.getRegistryGroup(), registryParam.getRegistryKey(),
                registryParam.getRegistryValue());
        if (ret > 0) {
          // fresh
          freshGroupRegistryInfo(registryParam);
        }
      }
    });

    return ReturnT.SUCCESS;
  }

  private void freshGroupRegistryInfo(RegistryParam registryParam) {
    // Under consideration, prevent affecting core tables
  }


}
